package com.fy.fyspace;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import org.junit.jupiter.api.Test;

import java.net.InetSocketAddress;

public class CanalUserTableWatcher {
    @Test
    public void watchUserTableChange() throws Exception {
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("192.168.131.128", 11111),
                "example", "", "");
        connector.connect();
        connector.subscribe("fyspace.user"); // 只监听user表
        connector.rollback();

        System.out.println("[CanalUserTableWatcher] Canal client started. Waiting for user table binlog...");

        for (int i = 0; i < 10; i++) {
            Message message = connector.getWithoutAck(100);
            long batchId = message.getId();
            int size = message.getEntries().size();
            if (batchId != -1 && size > 0) {
                message.getEntries().forEach(entry -> {
                    String entryStr = entry.toString();
                    // 简单判断操作类型
                    if (entryStr.contains("eventType: INSERT")) {
                        System.out.println("[user表插入] " + entryStr);
                    } else if (entryStr.contains("eventType: UPDATE")) {
                        System.out.println("[user表修改] " + entryStr);
                    } else if (entryStr.contains("eventType: DELETE")) {
                        System.out.println("[user表删除] " + entryStr);
                    }
                });
            } else {
                System.out.println("[CanalUserTableWatcher] No user table binlog entries. Try修改user表数据...");
            }
            connector.ack(batchId);
            Thread.sleep(1000);
        }
        connector.disconnect();
    }
} 